
 /**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017, chinazyjr All Rights Reserved.
 */
package com.sys.kafka.core.producer;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.sys.kafka.config.KafkaConfig;
import com.sys.kafka.config.KafkaConfig4Producer;
import com.sys.kafka.config.KafkaConfigHolder;
import com.sys.kafka.config.KafkaConfigUtil;
import com.sys.kafka.core.producer.callback.KCallback;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * <b>Description：</b> TODO <br/>
 * <b>ClassName：</b> ProducerHandler <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2017年7月13日 下午2:52:04 <br/>
 * <b>@version: </b>  <br/>
 */
public class ProducerHandler implements IProducerHandler {
    @SuppressWarnings("unused")
    private static final  Logger logger = LoggerFactory.getLogger(ProducerHandler.class);
    
    private  KafkaProducer<String, String> producer;
    
    public ProducerHandler(){
        
    }
    private final Properties props = new Properties();
    public void init(){
        KafkaConfigUtil.initConfig();
        KafkaConfig kafkaConfig= KafkaConfigHolder.get();
        //读取配置文件
        KafkaConfig4Producer kafkaConfig4Producer= kafkaConfig.getConfig4producer();
        props.put("bootstrap.servers", kafkaConfig.getBootstrapServers());
        props.put("acks", kafkaConfig4Producer.getAcks());
        props.put("retries",kafkaConfig4Producer.getRetries());
        props.put("batch.size", kafkaConfig4Producer.getBatch_size());
        props.put("linger.ms", kafkaConfig4Producer.getLinger_ms());
        props.put("buffer.memory", kafkaConfig4Producer.getBuffer_memory());
        props.put("key.serializer", kafkaConfig4Producer.getKey_serializer());
        props.put("value.serializer", kafkaConfig4Producer.getValue_serializer());
        producer = new KafkaProducer<String, String>(props);
    }
    
    @Override
    public void send(String topic,String msg) {
        
        send(topic,null,msg,null);
        
    }

    @Override
    public void send(String topic, String msg, KCallback callback) {
        send(topic,null,msg,callback);
        
    }

    @Override
    public void send(String topic, String key, String msg) {
        send(topic,key,msg,null);
    }

    @Override
    public void send(String topic, String key, String msg, KCallback callback) {
        if(producer==null){
            init();
        }
        
        ProducerRecord<String, String>producerRecord= new ProducerRecord<String, String>(topic,key, msg);
        producer.send(producerRecord,callback);
    }

}
